package com.hxq.demo5;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**生产者事务
 * @author hxq
 * @date 2022/4/21 14:50
 */
public class TransactionProducer {

    public static void main(String[] args) {
        haveCallback();
    }



    private static void haveCallback(){

        Properties properties = setConfigInfo();
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        // 初始化事务
        producer.initTransactions();
        // 开启事务
        producer.beginTransaction();

        try {
            //发送数据
            for (int i= 1; i<=4; i++){
                ProducerRecord<String, String> record = new ProducerRecord<>("first",
                        Integer.toString(i), "xxxxxxxxxx");
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e==null){
                            System.out.println("分区："+recordMetadata.partition()+"=====" + "游标："+recordMetadata.offset());
                        }else {
                            e.printStackTrace();
                        }

                    }
                });


            }
            // 提交事务
            producer.commitTransaction();
        }catch (Exception e){
            // 终止事务
            producer.abortTransaction();
        }finally {
            // 关闭资源
            producer.close();
        }


    }


    private static Properties setConfigInfo(){
        Properties properties = new Properties();
        //kafka集群
        properties.put("bootstrap.servers","212.129.155.35:9092");
        //ack
        properties.put("ack","all");

        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        // 设置事务 id（必须），事务 id 任意起名
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
                "transaction_id_0");


        return properties;

    }
}
